package cn.songtx.iot.config;

import lombok.Data;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

@Data
@Configuration
// 引入yml配置文件路径
@ConfigurationProperties(MqttConfig.PREFIX)
public class MqttConfig {

    /**
     * 1.获取连接参数
     */
    public static final String PREFIX = "spring.mqtt";  // 参数前缀
    private String host;                                // 创建的实体类和yml文件的配置项一致
    private String subClientId;
    private String subTopic;
    private String userName;
    private String password;
    private Integer timeout;
    private Integer keepAlive;
    private String pubClientId;

    /**
     * 2.构造一个options连接对象,并写入连接参数
     *
     * @return 返回连接对象
     */
    @Bean
    public MqttConnectOptions options() {
        MqttConnectOptions options = new MqttConnectOptions();
        // 如果设置为 false，客户端和服务器将在客户端、服务器和连接重新启动时保持状态。随着状态的保持：
        // 即使客户端、服务器或连接重新启动，消息传递也将可靠地满足指定的 QOS。服务器将订阅视为持久的。
        // 如果设置为 true，客户端和服务器将不会在客户端、服务器或连接重新启动时保持状态。
        options.setCleanSession(false);
        // 该值以秒为单位，必须>0，定义了客户端等待与 MQTT 服务器建立网络连接的最大时间间隔。
        // 默认超时为 30 秒。值 0 禁用超时处理，这意味着客户端将等待直到网络连接成功或失败。
        options.setConnectionTimeout(timeout);
        //此值以秒为单位，定义发送或接收消息之间的最大时间间隔，必须>0
        options.setKeepAliveInterval(keepAlive);
        //自动重新连接
        options.setAutomaticReconnect(true);
        options.setUserName(userName);
        options.setPassword(password.toCharArray());
        options.setServerURIs(new String[]{host});
        return options;
    }

    /**
     * 3.用上面的option()方法构造的连接参数注册MQTT客户端连接工厂
     *
     * @return 返回连接工厂
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory mqttPahoClientFactory = new DefaultMqttPahoClientFactory();
        mqttPahoClientFactory.setConnectionOptions(options());
        return mqttPahoClientFactory;
    }

    /**
     * 4.配置入站的消息通道
     *
     * @return 返回消息通道
     */
    @Bean
    public MessageChannel messageInputChannel() {
        return new DirectChannel();
    }

    /**
     * 5.配置入站的消息生产者,设置(clientId,连接工厂,订阅主题)
     *
     * @param mqttInputChannel  消息通道
     * @param mqttClientFactory 连接工厂
     * @return 返回消息生产者
     */
    @Bean
    public MessageProducer inbound(@Qualifier("messageInputChannel") MessageChannel mqttInputChannel, MqttPahoClientFactory mqttClientFactory) {
        // 构造入站适配器,设置(clientId,连接工厂,订阅主题)
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(subClientId, mqttClientFactory,
                        subTopic);
        adapter.setCompletionTimeout(8000);
        // 设置消息转换器
        adapter.setConverter(new DefaultPahoMessageConverter());
        // 设置Qos
        adapter.setQos(1);
        // 设置入站的通道(上面创建的消息通道)
        adapter.setOutputChannel(mqttInputChannel);
        return adapter;
    }

    /**
     * 6.配置出站的消息通道
     *
     * @return 返回消息通道
     */
    @Bean(value = "mqttOutboundChannel")
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * 7.配置出站的消息处理器,设置(clientId,连接工厂)
     *
     * @return 返回消息处理器
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        final MqttPahoMessageHandler handler = new MqttPahoMessageHandler(pubClientId, mqttClientFactory());
        handler.setDefaultQos(1);
        handler.setDefaultRetained(false);
        handler.setDefaultTopic(subTopic);
        handler.setAsync(false);
        handler.setAsyncEvents(false);
        return handler;
    }
}